package service

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"kafka-ui/backend/log"
	"kafka-ui/backend/types"
	"os"
	"strings"
	"sync"

	"github.com/jcmturner/gokrb5/v8/client"
	krbConfig "github.com/jcmturner/gokrb5/v8/config"
	"github.com/jcmturner/gokrb5/v8/keytab"
	"github.com/twmb/franz-go/pkg/kadm"
	"github.com/twmb/franz-go/pkg/kgo"
	"github.com/twmb/franz-go/pkg/sasl/kerberos"
	"github.com/twmb/franz-go/pkg/sasl/plain"
	"github.com/twmb/franz-go/pkg/sasl/scram"
)

type Service struct {
	connectName      string
	bootstrapServers []string
	config           []kgo.Opt
	kac              *kadm.Client
	client           *kgo.Client
	consumer         []any
	mutex            sync.Mutex
	topics           []any
	groups           []any
	sshTunnel        *sshTunnel
	log              *log.Logger
}

func NewKafkaService(l *log.Logger) *Service {
	return &Service{
		log: l,
	}
}

func (k *Service) Close(_ context.Context) {
	if k.client != nil {
		k.client.Close()
	}
	if k.kac != nil {
		k.kac.Close()
	}
	if len(k.consumer) == 2 {
		k.consumer[2].(*kgo.Client).Close()
	}
	if k.sshTunnel != nil {
		_ = k.sshTunnel.client.Close()
		k.sshTunnel = nil
	}
	fmt.Println("关闭 Kafka 服务连接")
}

// TestClient 测试连接
func (s *Service) TestClient(kafkaName string, conn map[string]any) *types.ResultResp {
	println("测试连接")
	return s.SetConnect(kafkaName, conn, true)
}

func (s *Service) SetConnect(kafkaName string, conn map[string]any, isTest bool) *types.ResultResp {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	result := &types.ResultResp{}

	var config []kgo.Opt

	// TLS配置
	if conn["tls"] == "enable" {
		tlsConfig := &tls.Config{
			InsecureSkipVerify: conn["skipTLSVerify"] == "true",
		}
		// 证书认证
		if conn["tlsCartFile"] != "" && conn["tlsKeyFile"] != "" {
			cert, err := tls.LoadX509KeyPair(conn["tlsCertFile"].(string), conn["tlsKeyFile"].(string))
			if err != nil {
				s.log.Info("加载客户端证书失败:", err)
				result.Err = fmt.Sprintf("加载客户端证书失败: %v", err)
				return result
			}
			tlsConfig.Certificates = []tls.Certificate{cert}
		}
		// CA认证
		if conn["tlsCAFile"] != "" {
			caCert, err := os.ReadFile(conn["tlsCAFile"].(string))
			if err != nil {
				s.log.Info("加载CA证书失败:", err)
				result.Err = fmt.Sprintf("加载CA证书失败: %v", err)
				return result
			}
			caCertPool := x509.NewCertPool()
			caCertPool.AppendCertsFromPEM(caCert)
			tlsConfig.RootCAs = caCertPool
		}
		config = append(config, kgo.DialTLSConfig(tlsConfig))
	}
	// SASL配置
	if conn["sasl"] == "enable" {
		user := conn["saslUser"].(string)
		password := conn["saslPassword"].(string)
		// SASL机制
		mechanism := conn["saslMechanism"].(string)
		switch strings.ToUpper(mechanism) {
		case "PLAIN":
			config = append(config, kgo.SASL(plain.Auth{User: user, Pass: password}.AsMechanism()))
		case "SCRAM-SHA-256":
			config = append(config, kgo.SASL(scram.Auth{User: user, Pass: password}.AsSha256Mechanism()))
		case "SCRAM-SHA-512":
			config = append(config, kgo.SASL(scram.Auth{User: user, Pass: password}.AsSha512Mechanism()))
		case "GSSAPI":
			kt, err := keytab.Load(conn["kerberosUserKeytab"].(string))
			if err != nil {
				result.Err = err.Error()
				return result
			}
			cfg, err := krbConfig.Load(conn["kerberosKrb5Conf"].(string))
			if err != nil {
				result.Err = err.Error()
				return result
			}
			// 创建客户端
			kerberosClient := client.NewWithKeytab(
				conn["kerberosUser"].(string),
				conn["kerberosRealm"].(string),
				kt,
				cfg,
				client.DisablePAFXFAST(true),
			)
			// 创建GSSAPI认证
			config = append(config, kgo.SASL(kerberos.Auth{
				Client:           kerberosClient,
				Service:          conn["kerberosServiceName"].(string),
				PersistAfterAuth: true,
			}.AsMechanism()))
		default:
			s.log.Info("不支持的 SASL 机制:", mechanism)
			result.Err = fmt.Sprintf("不支持的 SASL 机制: %s", mechanism)
			return result
		}
	}
	// 读取 kakfa 连接地址
	bootstrapServers := strings.Split(conn["bootstrapServers"].(string), ",")
	config = append(config, kgo.SeedBrokers(bootstrapServers...))
	cl, err := kgo.NewClient(config...)
	if err != nil {
		result.Err = "NewClient 失败: " + err.Error()
		return result
	}
	admin := kadm.NewClient(cl)
	ctx := context.Background()
	topics, err := admin.ListTopics(ctx)
	if err != nil {
		s.log.Info("连接 Kafka 失败:", err)
		result.Err = "连接 Kafka 失败: " + err.Error()
		return result
	}
	//正式切换节点，赋值并清理缓存，并更新缓存
	if !isTest {
		s.connectName = kafkaName
		s.kac = admin
		s.client = cl
		s.bootstrapServers = bootstrapServers
		s.clearCache()
		s.topics = s.buildTopicsResp(topics)
	}
	return result
}

func (s *Service) buildTopicsResp(topics kadm.TopicDetails) []any {
	var result []any
	for topicName, topicDetail := range topics {
		partitionErrs := ""
		var partitions []any
		for _, partition := range topicDetail.Partitions {
			errMsg := ""
			if partition.Err != nil {
				errMsg = partition.Err.Error()
				partitionErrs += fmt.Sprintf("partition %d: %s\n", partition.Partition, errMsg)
			}
			partitions = append(partitions, map[string]any{
				"partition":       partition.Partition,
				"leader":          partition.Leader,
				"replicas":        partition.Replicas,
				"isr":             partition.ISR,
				"err":             errMsg,
				"LeaderEpoch":     partition.LeaderEpoch,
				"OfflineReplicas": partition.OfflineReplicas,
			})
		}
		if topicDetail.Err != nil {
			partitionErrs = topicDetail.Err.Error() + "\n" + partitionErrs
		}
		// 检查分区列表是否为空，避免访问空切片的第一个元素
		replicationFactor := 0
		if len(topicDetail.Partitions) > 0 {
			replicationFactor = len(topicDetail.Partitions[0].Replicas)
		}
		result = append(result, map[string]any{
			"ID":                 topicDetail.ID,
			"topic":              topicName,
			"partition_count":    len(topicDetail.Partitions),
			"replication_factor": replicationFactor,
			"IsInternal":         topicDetail.IsInternal,
			"Err":                partitionErrs,
			"partitions":         partitions,
		})
	}
	return result
}

func (s *Service) clearCache() {
	s.topics = nil
	s.groups = nil
}
